#ifndef __REPLICA_H__
#define __REPLICA_H__

#include <stdint.h>

#include "job.h"
#include "chunk.h"
#include "cache.h"
#include "diskmd.h"
#include "buffer.h"
#include "clock/clock.h"
#include "../lease/lease_ctl.h"
#include "bit_lock.h"

typedef struct {
        chkid_t chkid;
        chkid_t parent;
        diskloc_t loc;
        nid_t owner;
        lease_token_t token;
        uint32_t magic;
        
#if ENABLE_CHUNK_PARALLEL
        struct __pio {
                sy_spinlock_t lock;
                uint32_t writing;
                bits_lock_t *bits_lock;
        } pio;
#endif

        vclock_t vclock;
        struct __wq {
                sy_spinlock_t lock;
                struct list_head wlist;   ///< wait list
        } wq;

        char *pool;

        //to be removed
        //uint64_t fingerprint;
        //int wbdisk;
} replica_srv_entry_t;

typedef struct {
        char home[MAX_PATH_LEN];
} replica_srv_t;

typedef struct {
        vclock_t vclock;
        uint16_t dirty;
        uint16_t lost;
} clockstat_t;

typedef enum {
        __OP_NULL__,
        __OP_READ_WAIT__,
        __OP_WRITE_WAIT__,
        __OP_WRITING__,
        __OP_READING__,
} wlist_op_t;

typedef struct {
        struct list_head hook;
        char op;
        chkid_t chkid;
        nid_t writer;
        char committed;
        vclock_t vclock;
        time_t begin;
        time_t ltime;
        uint32_t magic;
        task_t task;
} wlist_t;

#define REPLICA_MAX_RETRY 150

#define REPLICA_DISK_MD_PAGE_SIZE       4096

// local or RPC wrapper
int replica_diskonline(const chkinfo_t *chkinfo, int idx);

/*from replica_rpc.c*/

int replica_rpc_init(void);

int replica_rpc_create(const char *pool, const nid_t *nid, const chkid_t *chkid,
                       int chknum, const fileid_t *parent, int tier, int initzero, const buffer_t *buf,
                       uint64_t meta_version, int force);
int replica_rpc_unlink(const nid_t *nid, const chkid_t *chkid, uint64_t meta_version);
int replica_rpc_cleanup(const nid_t *nid, const chkid_t *chkid);

int replica_rpc_push(const char *pool, const nid_t *nid, const chkid_t *chkid, const fileid_t *parent,
                     int tier, const vclock_t *vclock, uint64_t meta_version, const buffer_t *_buf, int flags);

int replica_rpc_connect(const nid_t *nid, const chkid_t *chkid, const chkid_t *parent,
                        const lease_token_t *token, uint32_t magic, clockstat_t *clockstat, int force);

int replica_rpc_diskonline(const nid_t *nid, const chkid_t *chkid, int *online);

int replica_rpc_sha1(const nid_t *nid, const chkid_t *chkid, uint64_t version, char *buf);

int replica_rpc_getclock(const nid_t *nid, const chkid_t *chkid, clockstat_t *clockstat);
int replica_rpc_setclock(const nid_t *nid, const chkid_t *chkid, const clockstat_t *clockstat);

int replica_rpc_getparent(const diskid_t *nid, const chkid_t *chkid, chkid_t *parent);
int replica_rpc_setparent(const diskid_t *nid, const chkid_t *chkid, const chkid_t *parent);

int replica_rpc_settier(const nid_t *nid, const chkid_t *chkid, int num);
int replica_rpc_gettier(const nid_t *nid, const chkid_t *chkid, int *num);

int replica_rpc_setpriority(const nid_t *nid, const chkid_t *chkid, int num);
int replica_rpc_getpriority(const nid_t *nid, const chkid_t *chkid, int *num);

int replica_rpc_read(const diskid_t *diskid, const io_t *io, buffer_t *buf);
int replica_rpc_write(const diskid_t *disk, const io_t *io, const buffer_t *buf, uint32_t magic);

int replica_rpc_analysis_update(const nid_t *nid, const chkid_t *chkid, uint32_t read, uint32_t write);

/*from replica_srv.c*/

int replica_srv_prep();
int replica_srv_init(const char *home, uint64_t _max_object);
void replica_srv_destroy();

int replica_srv_get(const chkid_t *id, mcache_entry_t **_cent);
int replica_srv_release(mcache_entry_t *cent);

// 改变sqlite和disk bitmap的操作 {{

// TODO 两个并发的删除副本的过程，需要进行同步。否则，会造成严重后果。

// case 1: 两个并发任务都从sqlite读出了同一个chunk，一个并发任务清除disk bitmap后，被reuse，
// 另一并发任务此时再次清除该disk bitmap(包含有效数据）。

// case 2: sqlite有，而disk bitmap无，后续会造成两个sqlite记录指向同一磁盘位置

// case 3: msqqueue没有回收，造成chunk实际副本数多于chkinfo所指定的副本集合

int replica_srv_create(const char *pool, const nid_t *master, const chkid_t *chkids, int chknum,
                       const fileid_t *parent, int tier, int initzero,
                       const buffer_t *initdata, uint64_t meta_version, int force);
int replica_srv_create_with_fingerprint(const char *pool, const nid_t *owner,
                                        const chkid_t *chkid, int chknum, const fileid_t *parent, int tier, int initzero,
                                        const buffer_t *initdata, uint64_t meta_version, uint64_t fingerprint, int force);

int replica_srv_unlink(const chkid_t *chkid, uint64_t meta_version);

int replica_srv_cleanup(const chkid_t *chkid, int clean_md);

int replica_srv_push(const char *pool, const nid_t *owner, const chkid_t *chkid, const fileid_t *parent, int tier,
                     const vclock_t *vclock, uint64_t meta_version, const buffer_t *buf, int flags);

// 删除disk时调用
int replica_srv_cast(const chkid_t *id, uint32_t from);

// }}

// 查询状态

int replica_srv_connect(const nid_t *owner, const chkid_t *chkid,
                        const chkid_t *parent, const lease_token_t *token, uint32_t magic,
                        clockstat_t *clockstat, int force);
int replica_srv_diskonline(const chkid_t *chkid, int *online);
int replica_srv_getinfo(const chkid_t *chkid, diskloc_t *loc, chkid_t *parent,
                        uint64_t *fingerprint, int *wbdisk);
int replica_srv_sha1(const chkid_t *chkid, uint64_t version, char *buf);

int replica_srv_getclock(const nid_t *master, const chkid_t *chkid, clockstat_t *clockstat);
int replica_srv_setclock(const nid_t *owner, const chkid_t *chkid, const clockstat_t *clockstat);

int replica_srv_getparent(const chkid_t *chkid, chkid_t *parent, char *pool);
int replica_srv_setparent(const chkid_t *chkid, const chkid_t *parent);

// IO

int replica_srv_read(const nid_t *reader, const io_t *io, buffer_t *buf);
int replica_srv_sync_read(const chkid_t *chkid, void *buf);
int replica_srv_write(const nid_t *writer, const io_t *io, const buffer_t *buf, uint32_t magic);

int replica_cleanup_init();

int replica_srv_dump_memory(uint64_t *memory);

inline static int __replica_srv_wqcheck(replica_srv_entry_t *ent)
{
        int ret;

        ret = sy_spin_lock(&ent->wq.lock);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (!list_empty(&ent->wq.wlist)) {
                ret = EBUSY;
                GOTO(err_lock, ret);
        }

        sy_spin_unlock(&ent->wq.lock);

        return 0;
err_lock:
        sy_spin_unlock(&ent->wq.lock);
err_ret:
        return ret;
}

inline static void __replica_srv_wqadd(replica_srv_entry_t *ent, wlist_t *list)
{
        int ret;

        ret = sy_spin_lock(&ent->wq.lock);
        if (unlikely(ret))
                UNIMPLEMENTED(__DUMP__);

        list_add_tail(&list->hook, &ent->wq.wlist);

        sy_spin_unlock(&ent->wq.lock);
}

#endif
